package com.atguigu.api

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @ClassName ProcessFunctionTest2
 * @Description
 * @Author Mr Yang
 * @Date 2020/9/9 21:11
 * @Version 1.0
 */
object ProcessFunctionTest2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //开启检查点
    env.enableCheckpointing(60000)

    val stream = env.socketTextStream("localhost",7777)

    val dataStream = stream.map(data=>{
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
          override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
        })

    val processStream = dataStream.keyBy(_.id)
        //.process(new TempChangeAlert(10.0))
        .flatMap( new TempChangeAlert(10.0))

    val processStream2 = dataStream.keyBy(_.id)
        .flatMapWithState[(String, Double, Double), Double]{
          //如果没有状态的话，也就没有数据来过，那么就将当前输入的温度值存入状态
          case (input: SensorReading, None) => (List.empty, Some(input.temperature))
            //如果有状态，就应该与上次的温度值比较差值，如果大于 阈值就输出报警
          case (input: SensorReading, lastTempState: Some[Double]) =>{
            val diff = (input.temperature - lastTempState.get).abs
            if (diff > 10.0){
              (List((input.id, lastTempState.get, input.temperature)), Some(input.temperature))
            }else{
              (List.empty, Some(input.temperature))
            }
          }
        }


    processStream2.print("processStream2 test")
    dataStream.print("dataStream test")
    processStream.print("processStream test")

    env.execute("温度连续变化大于10 test")
  }
}

class TempChangeAlert(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)]{

  private var lastTempState: ValueState[Double] = _

  override def open(parameters: Configuration): Unit = {
    //初始化的时候声明变量state
    lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  }

  override def flatMap(in: SensorReading, collector: Collector[(String, Double, Double)]): Unit = {
    //获取上次的温度值
    val lastTemp = lastTempState.value()
    //当前温度值与上次温度值计算大于10，输出报警信息
    val diff = (in.temperature - lastTemp).abs   //abs 绝对值

    if (diff > threshold){
      collector.collect((in.id, lastTemp, in.temperature))
    }
    //更新状态信息
    lastTempState.update(in.temperature)
  }
}

//输出id，两次温度的变化
class TempChangeAlert2(threshold: Double) extends KeyedProcessFunction[String, SensorReading,(String, Double, Double)]{
  //定义一个状态变量，保存上次的温度值
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))

  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, (String, Double, Double)]#Context, collector: Collector[(String, Double, Double)]): Unit = {
    //获取上次的温度值
    val lastTemp = lastTempState.value()
    //当前温度值与上次温度值计算大于10，输出报警信息
    val diff = (i.temperature - lastTemp).abs   //abs 绝对值

    if (diff > threshold){
      collector.collect((i.id, lastTemp, i.temperature))
    }
    //更新状态信息
    lastTempState.update(i.temperature)
  }
}